#ifndef UL_CONDITION_TASK_QUEUE_H_ 
#define UL_CONDITION_TASK_QUEUE_H_
#include "ul_task_queue.h" 
#include <pthread.h>
#include <sys/time.h>
#include "ul_link.h"
namespace LinuxNetworkUtility
{
  class UlConditionQueenWorkerTQ
  {
  public:
    UlConditionQueenWorkerTQ();
    virtual ~UlConditionQueenWorkerTQ();

    virtual void* malloc_tsk(const int64_t size, const int32_t mod_id);

    virtual void free_tsk(void* tsk); 

    virtual int init(const int64_t worker_bee_num);
    /// the queen bee dispatch task to worker bees
    virtual int dispatch_task(void *tsk);
    /// the worker bee claim task to do
    virtual int claim_task(void **tsk, const int64_t ntsk, 
                           const int64_t timeout_us, const int64_t worker_id);
    /// the worker bee submit output of task
    virtual int submit_output(void *tsk, const int64_t worker_id);
    /// the queen bee fetch output of tasks
    virtual int fetch_output(void **tsk, const int64_t ntsk, const int64_t timeout_us);
  private:
    void clear();
    struct UlTsk
    {
      uint64_t magic_;
      DLink link_;
      char  tsk_[0];
    };
    struct UlConditionPipe
    {
      int64_t pending_tsk_num_;
      DLink worker_end_list_;
      DLink queen_end_list_;
      pthread_mutex_t     mutex_;
      pthread_cond_t      cond_;
      pthread_spinlock_t  spin_; 
      UlConditionPipe()
      {
        pthread_mutex_init(&(mutex_),NULL);
        pthread_cond_init(&(cond_),NULL);
        pthread_spin_init(&(spin_), PTHREAD_PROCESS_PRIVATE);
        pending_tsk_num_ = 0;
      }

      inline void enqueue_task(UlTsk * tsk)
      {
        pthread_mutex_lock(&(mutex_));
        worker_end_list_.insert_prev(tsk->link_);
        pending_tsk_num_ ++;
        pthread_cond_signal(&(cond_));
        pthread_mutex_unlock(&(mutex_));
      }

      inline int64_t claim_task(void **tsk, const int64_t ntsk, const int64_t timeout_us)
      {
        DLink *it = NULL;
        UlTsk *ltsk = NULL;
        int64_t got_tsk_num = 0;
        UNUSED(timeout_us);
        struct timeval now;
        struct timespec timeout;
        if (timeout_us > 0)
        {
          gettimeofday(&now,NULL);
          int64_t dst_time_us = timeout_us + now.tv_usec;
          timeout.tv_sec = now.tv_sec + dst_time_us/1000000;
          timeout.tv_nsec = (dst_time_us%1000000)*1000;
        }
        pthread_mutex_lock(&(mutex_));
        if (worker_end_list_.is_empty())
        {
          if (timeout_us > 0)
          {
            pthread_cond_timedwait(&cond_, &mutex_, &timeout);
          }
          else
          {
            pthread_cond_wait(&cond_,&mutex_);
          }
        }
        it = worker_end_list_.next();
        while (got_tsk_num < ntsk && it != &worker_end_list_)
        {
          ltsk = CONTAINING_RECORD(it, UlTsk, link_);
          it = it->next();
          ltsk->link_.remove();
          tsk[got_tsk_num] = ltsk->tsk_;
          got_tsk_num ++;
        }
        pthread_mutex_unlock(&(mutex_));
        if (0 >= got_tsk_num)
        {
          errno = EAGAIN;
          got_tsk_num = ERRNO2ERR(errno);
        }
        return got_tsk_num;
      }

      inline void submit_output(UlTsk *tsk)
      {
        pthread_spin_lock(&spin_);
        queen_end_list_.insert_prev(tsk->link_);
        pthread_spin_unlock(&spin_);
      }

      inline int64_t fetch_output(void **tsk, const int64_t ntsk)
      {
        DLink *it = NULL;
        UlTsk *ltsk = NULL;
        int64_t got_tsk_num = 0;
        pthread_spin_lock(&spin_);
        it = queen_end_list_.next();
        while (got_tsk_num < ntsk && it != &queen_end_list_)
        {
          ltsk = CONTAINING_RECORD(it, UlTsk, link_);
          it = it->next();
          ltsk->link_.remove();
          tsk[got_tsk_num] = ltsk->tsk_;
          got_tsk_num ++;
        }
        pthread_spin_unlock(&spin_);
        return got_tsk_num;
      }
    };

    int64_t     next_push_pipe_idx_;
    int64_t     worker_num_;
    UlConditionPipe     *cond_pipes_;
  };
}

#endif /* UL_CONDITION_TASK_QUEUE_H_ */
